package top.luoyuanxiang.mqtt.consumer;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import top.luoyuanxiang.mqtt.kafka.KafkaProducer;

/**
 * 订阅主题消息处理
 *
 * @author luoyuanxiang
 */
@Slf4j
@Configuration
public class MqttMessageConsumer {

    @Resource
    private KafkaProducer kafkaProducer;

    /**
     * 接收消息
     *
     * @param message 消息内容
     */
    @ServiceActivator(inputChannel = "mqtt5InboundChannel")
    public void handleMessage(Message<byte[]> message) {
        MessageHeaders headers = message.getHeaders();
        // 获取消息Topic
        String receivedTopic = (String) headers.get(MqttHeaders.RECEIVED_TOPIC);
        // 存入kafka
        kafkaProducer.sendMessage(receivedTopic, new String(message.getPayload()));
        // 消息处理
        log.info("received topic: {}, message: {}", receivedTopic, new String(message.getPayload()));
        // 根据消息ID处理

    }
}
